package com.wa.service;

import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.allchips.common.RestAPICode;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.SocketIONamespace;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.protocol.Packet;
import com.corundumstudio.socketio.protocol.PacketType;
import com.wa.allchips.util.CommUtils;
import com.wa.base.GenericResp;
import com.wa.config.WebSocketConfig;
import com.wa.constants.NameSpaceBeanNameConstant;
import com.wa.constants.WebSocketEventConstant;
import com.wa.enums.NameSpaceEnum;
import com.wa.util.WebSocketSessionKeyUtil;
import com.wa.util.WebSocketSessionManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

@Service
@ConditionalOnBean(value = WebSocketConfig.class)
public class WebSocketPushApiImpl{

    final static private Logger logger = LoggerFactory.getLogger(WebSocketPushApiImpl.class);

    @Resource
    private SocketIOServer socketIOServer;

    @Resource(name = NameSpaceBeanNameConstant.BOM)
    private SocketIONamespace bomSocketIONamespace;

    @Resource(name = NameSpaceBeanNameConstant.BOSS)
    private SocketIONamespace bossSocketIONamespace;

    @Resource
    private WebSocketSessionManager webSocketSessionManager;



    /**
     * 推送至指定websocket(从协同工具连接)
     *
     * @param uuids   socket uuid
     * @param message data
     * @return
     */
    @SuppressWarnings("deprecation")
    public GenericResp<List<GenericResp<Boolean>>> push2WebSocketFromBom(List<String> uuids, Object message) {
        GenericResp<List<GenericResp<Boolean>>> returnVo = new GenericResp<>();
        try {
            returnVo.setData(uuids.stream().map(uuid -> doPush(uuid, message, WebSocketEventConstant.MESSAGE_EVENT, getSocketClient(bomSocketIONamespace, uuid))).collect(Collectors.toList()));
        } catch (Exception e) {
            logger.error("推送异常", e);
            returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        }
        return returnVo;
    }

    /**
     * 推送至指定websocket(从协同工具连接)
     *
     * @param uuids
     * @param json
     * @return
     */
    public GenericResp<List<GenericResp<Boolean>>> push2WebSocketFromBom(List<String> uuids, String json) {
        GenericResp<List<GenericResp<Boolean>>> returnVo = new GenericResp<>();
        try {
            returnVo.setData(uuids.stream().map(uuid -> doPush(uuid, json, WebSocketEventConstant.MESSAGE_EVENT, getSocketClient(bomSocketIONamespace, uuid))).collect(Collectors.toList()));
        } catch (Exception e) {
            logger.error("推送异常", e);
            returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        }
        return returnVo;
    }


    /**
     * 推送至指定后端用户(从协同工具连接)
     *
     * @param userIds backend user id
     * @param message data
     * @return
     */
    @SuppressWarnings("deprecation")
    public GenericResp<List<GenericResp<Boolean>>> push2backendUserFromBom(List<String> userIds, Object message) {
        GenericResp<List<GenericResp<Boolean>>> returnVo = new GenericResp<>();
        try {
            returnVo.setData(userIds.stream().map(userId -> doPush(userId, message, WebSocketEventConstant.MESSAGE_EVENT, getUserSocketClient(bomSocketIONamespace, WebSocketSessionKeyUtil.buildWebsiteBackendBomKey(userId)))).collect(Collectors.toList()));
        } catch (Exception e) {
            logger.error("推送异常", e);
            returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        }
        return returnVo;
    }

    /**
     * 推送至指定后端用户(从协同工具连接)
     *
     * @param userIds
     * @param json
     * @return
     */
    public GenericResp<List<GenericResp<Boolean>>> push2backendUserFromBom(List<String> userIds, String json) {
        GenericResp<List<GenericResp<Boolean>>> returnVo = new GenericResp<>();
        try {
            returnVo.setData(userIds.stream().map(userId -> doPush(userId, json, WebSocketEventConstant.MESSAGE_EVENT, getUserSocketClient(bomSocketIONamespace, WebSocketSessionKeyUtil.buildWebsiteBackendBomKey(userId)))).collect(Collectors.toList()));
        } catch (Exception e) {
            logger.error("推送异常", e);
            returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        }
        return returnVo;
    }


    /**
     * 推送至指定websocket(从boss后端连接)
     *
     * @param uuids   socket uuid
     * @param message data
     * @return
     */
    @SuppressWarnings("deprecation")
    public GenericResp<List<GenericResp<Boolean>>> push2WebSocketFromBoss(List<String> uuids, Object message) {
        GenericResp<List<GenericResp<Boolean>>> returnVo = new GenericResp<>();
        try {
            returnVo.setData(uuids.stream().map(uuid -> doPush(uuid, message, WebSocketEventConstant.MESSAGE_EVENT, getSocketClient(bossSocketIONamespace, uuid))).collect(Collectors.toList()));
        } catch (Exception e) {
            logger.error("推送异常", e);
            returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        }
        return returnVo;
    }

    /**
     * 推送至指定websocket(从boss后端连接)
     *
     * @param uuids
     * @param json
     * @return
     */
    public GenericResp<List<GenericResp<Boolean>>> push2WebSocketFromBoss(List<String> uuids, String json) {
        GenericResp<List<GenericResp<Boolean>>> returnVo = new GenericResp<>();
        try {
            returnVo.setData(uuids.stream().map(uuid -> doPush(uuid, json, WebSocketEventConstant.MESSAGE_EVENT, getSocketClient(bossSocketIONamespace, uuid))).collect(Collectors.toList()));
        } catch (Exception e) {
            logger.error("推送异常", e);
            returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        }
        return returnVo;
    }


    /**
     * 推送至指定后端用户(从boss后端连接)
     *
     * @param userIds
     * @param message
     * @return
     */
    @SuppressWarnings("deprecation")
    public GenericResp<List<GenericResp<Boolean>>> push2backendUserFromBoss(List<String> userIds, Object message) {
        GenericResp<List<GenericResp<Boolean>>> returnVo = new GenericResp<>();
        try {
            returnVo.setData(userIds.stream().map(userId -> doPush(userId, message, WebSocketEventConstant.MESSAGE_EVENT, getUserSocketClient(bossSocketIONamespace, WebSocketSessionKeyUtil.buildWebsiteBackendBossKey(userId)))).collect(Collectors.toList()));
        } catch (Exception e) {
            logger.error("推送异常", e);
            returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        }
        return returnVo;
    }

    /**
     * 推送至指定后端用户(从boss后端连接)
     *
     * @param userIds
     * @param json
     * @return
     */
    public GenericResp<List<GenericResp<Boolean>>> push2backendUserFromBoss(List<String> userIds, String json) {
        GenericResp<List<GenericResp<Boolean>>> returnVo = new GenericResp<>();
        try {
            returnVo.setData(userIds.stream().map(userId -> doPush(userId, json, WebSocketEventConstant.MESSAGE_EVENT, getUserSocketClient(bossSocketIONamespace, WebSocketSessionKeyUtil.buildWebsiteBackendBossKey(userId)))).collect(Collectors.toList()));
        } catch (Exception e) {
            logger.error("推送异常", e);
            returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        }
        return returnVo;
    }

    /**
     * 指定推送, 前端将收到json|string对象
     *
     * @param uuid
     * @param str
     * @return
     */
    public GenericResp<Boolean> push2WebSocket(int service, String uuid, String str) {
        GenericResp<Boolean> returnVo = new GenericResp<>();
        try {
            SocketIONamespace namespace = null;
            if (CommUtils.isNotNull(NameSpaceEnum.get(service))) {
                namespace = SpringUtil.getBean(NameSpaceEnum.get(service).getCodeCn() + NameSpaceBeanNameConstant.SUFFIX);
            }
            if (null != namespace) {
                return doPush(uuid, str, WebSocketEventConstant.MESSAGE_EVENT, getSocketClient(namespace, uuid));
            } else {
                returnVo.setMessage("命名空间不存在");
                returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
            }
        } catch (Exception e) {
            logger.error("推送异常", e);
            returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        }
        return returnVo;
    }

    /**
     * 指定推送, 前端将收到json|string对象
     *
     * @param uuids
     * @param str
     * @return
     */
    public GenericResp<List<GenericResp<Boolean>>> push2WebSocket(int service, List<String> uuids, String str) {
        GenericResp<List<GenericResp<Boolean>>> returnVo = new GenericResp<>();
        try {
            SocketIONamespace namespace = null;
            if (CommUtils.isNotNull(NameSpaceEnum.get(service))) {
                namespace = SpringUtil.getBean(NameSpaceEnum.get(service).getCodeCn() + NameSpaceBeanNameConstant.SUFFIX);
            }
            final SocketIONamespace finalNamespace = namespace;
            if (null != finalNamespace) {
                returnVo.setData(uuids.stream().map(uuid -> doPush(uuid, str, WebSocketEventConstant.MESSAGE_EVENT, getSocketClient(finalNamespace, uuid))).collect(Collectors.toList()));
            } else {
                returnVo.setMessage("命名空间不存在");
                returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
            }
        } catch (Exception e) {
            logger.error("推送异常", e);
            returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        }
        return returnVo;
    }

    /**
     * 是否在线
     *
     * @param service
     * @param uuid
     * @return
     */
    public GenericResp<Boolean> online(int service, String uuid) {
        GenericResp<Boolean> returnVo = new GenericResp<>();
        try {
            SocketIONamespace namespace = null;
            if (CommUtils.isNotNull(NameSpaceEnum.get(service))) {
                namespace = SpringUtil.getBean(NameSpaceEnum.get(service).getCodeCn() + NameSpaceBeanNameConstant.SUFFIX);
            }
            final SocketIONamespace finalNamespace = namespace;
            if (null != finalNamespace) {
                returnVo.setData(getSocketClient(finalNamespace, uuid) == null ? Boolean.FALSE: Boolean.TRUE);
            } else {
                returnVo.setMessage("命名空间不存在");
                returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
            }
        } catch (Exception e) {
            logger.error("推送异常", e);
            returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        }
        return returnVo;
    }

    /**
     *
     * @param socketIONamespace
     * @param uuid
     * @return
     */
    private SocketIOClient getSocketClient(SocketIONamespace socketIONamespace, String uuid) {
        UUID uid = null;
        try {
            uid = UUID.fromString(uuid);
        } catch (Exception e) {
            logger.debug("uid输入不正确");
        }
        SocketIOClient client;
        try {
            client = socketIONamespace.getClient(uid);
        } catch (Exception e) {
            return null;
        }
        // 命名空间没有的情况下, 从根空间获取, 会导致前端无法监听事件, 监听事件是跟命名空间绑定的
        if (null == client || !client.isChannelOpen()) {
            client = socketIOServer.getClient(UUID.fromString(uuid));
        }
        return client;
    }

    /**
     *
     * @param id
     * @param message
     * @param event
     * @param client
     * @return
     */
    private GenericResp<Boolean> doPush(String id, Object message, String event, SocketIOClient client) {
        if (null == client || !client.isChannelOpen()) {
            logger.info("用户不在线{id/uuid:{}}", id);
            return offlineReturnVo(id);
        }
        try {
            if (CommUtils.isEmpty(event)) {
                event = WebSocketEventConstant.MESSAGE_EVENT;
            }
            if (message instanceof String) {
                String copyMessage = (String) message;
                boolean decoded = true;
                try {
                    message = JSONObject.parseObject(copyMessage);
                } catch (Exception e) {
                    decoded = false;
                }
                if (!decoded) {
                    try {
                        decoded = true;
                        message = JSONArray.parseArray(copyMessage);
                    } catch (Exception e) {
                        decoded = false;
                    }
                }
                if (!decoded) {
                    logger.info("不是json串, 作为普通串发布");
                }
            }
            client.send(packet(event, message));
        } catch (Exception e) {
            logger.error("推送异常", e);
            return errorReturnVo(id);
        }
        return successReturnVo(id);
    }

    public Packet packet(String event, Object data) {
        Packet packet = new Packet(PacketType.MESSAGE);
        packet.setSubType(PacketType.EVENT);
        packet.setName(event);
        if (data instanceof List) {
            packet.setData(data);
        } else {
            packet.setData(Collections.singletonList(data));
        }
        return packet;
    }


    /**
     *
     * @param socketIONamespace
     * @param cacheKey
     * @return
     */
    private SocketIOClient getUserSocketClient(SocketIONamespace socketIONamespace, String cacheKey) {
        HashSet<UUID> set = webSocketSessionManager.getUserClient(cacheKey);
        if (CollectionUtils.isEmpty(set)) {
            logger.info("不在线cacheKey: {}", cacheKey);
            return null;
        }
        Iterator<UUID> it = set.iterator();
        SocketIOClient client = null;
        while (null == client && it.hasNext()) {
            final UUID uuid = it.next();
            try {
                client = socketIONamespace.getClient(uuid);
            } catch(Exception ignored) {
            }
        }
        // 命名空间没有的情况下, 从根空间获取, 会导致前端无法监听事件, 监听事件是跟命名空间绑定的
        if (null == client || !client.isChannelOpen()) {
            logger.debug("<-------------------------------------------开始从跟路径下获取客户端信息------------------------------------------->");
            it = set.iterator();
            while (null == client && it.hasNext()) {
                final UUID uuid = it.next();
                try {
                    client = socketIOServer.getClient(uuid);
                } catch (Exception ignored) {
                }
            }
        }
        return client;
    }

    static GenericResp<Boolean> successReturnVo(Object param) {
        GenericResp<Boolean> returnVo = new GenericResp<>();
        returnVo.setData(Boolean.TRUE);
        returnVo.setMessage(String.format("推送成功：%s", JSONUtil.toJsonStr(param)));
        return returnVo;
    }

    static GenericResp<Boolean> errorReturnVo(Object param) {
        GenericResp<Boolean> returnVo = new GenericResp<>();
        returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        returnVo.setData(Boolean.FALSE);
        returnVo.setMessage(String.format("推送异常: %s", JSONUtil.toJsonStr(param)));
        return returnVo;
    }

    static GenericResp<Boolean> offlineReturnVo(Object param) {
        GenericResp<Boolean> returnVo = new GenericResp<>();
        returnVo.setCode(RestAPICode.DATABASE_ERROR.getCode());
        returnVo.setData(Boolean.FALSE);
        returnVo.setMessage(String.format("未上线: %s", JSONUtil.toJsonStr(param)));
        return returnVo;
    }

}
